{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import broadcast, SparkConf\n",
    "import time\n",
    "import os\n",
    "\n",
    "RAPIDS_JAR = os.getenv(\"RAPIDS_JAR\", \"/path/to/your/jars/rapids.jar\")\n",
    "SPARK_MASTER = os.getenv(\"SPARK_MASTER_URL\", \"spark://ip:port\")\n",
    "print(\"RAPIDS_JAR: {}\".format(RAPIDS_JAR))\n",
    "if \"sc\" in globals():\n",
    "    sc.stop()\n",
    "\n",
    "### Configure the parameters based on your dataproc cluster ###\n",
    "conf = SparkConf().setAppName(\"Retail Analytics\")\n",
    "conf.setMaster(SPARK_MASTER)\n",
    "conf.set(\"spark.driver.extraClassPath\", RAPIDS_JAR)\n",
    "conf.set(\"spark.executor.extraClassPath\", RAPIDS_JAR)\n",
    "conf.set(\"spark.jars\", RAPIDS_JAR)\n",
    "conf.set(\"spark.executor.instances\", \"1\")\n",
    "conf.set(\"spark.executor.cores\", \"4\")\n",
    "conf.set(\"spark.task.resource.gpu.amount\", \"0.25\")\n",
    "conf.set(\"spark.rapids.sql.concurrentGpuTasks\", \"2\")\n",
    "conf.set(\"spark.executor.memory\", \"4g\")\n",
    "conf.set(\"spark.sql.files.maxPartitionBytes\", \"128m\")\n",
    "conf.set(\"spark.executor.resource.gpu.amount\", \"1\")\n",
    "conf.set(\"spark.rapids.memory.pinnedPool.size\", \"2048m\")\n",
    "conf.set(\"spark.executor.memoryOverhead\", \"4096m\")\n",
    "conf.set(\"spark.dynamicAllocation.enabled\", \"false\")\n",
    "conf.set(\"spark.rapids.sql.format.json.read.enabled\",True)\n",
    "conf.set(\"spark.rapids.sql.castStringToTimestamp.enabled\",True)\n",
    "conf.set(\"spark.rapids.sql.expression.PercentRank\",False)\n",
    "conf.set(\"spark.rapids.sql.castDecimalToString.enabled\",True)\n",
    "conf.set(\"spark.rapids.sql.hasExtendedYearValues\",False)\n",
    "conf.set(\"spark.rapids.sql.enabled\",True)\n",
    "conf.set(\"spark.plugins\", \"com.nvidia.spark.SQLPlugin\")\n",
    "conf.set(\"spark.rapids.sql.allowMultipleJars\", \"ALWAYS\")\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "                    .config(conf=conf) \\\n",
    "                    .getOrCreate()\n",
    "# create a SparkSession\n",
    "spark = SparkSession.builder.appName(\"RetailInvMgmt\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "# You need to update these to your real paths!\n",
    "dataRoot = os.getenv(\"DATA_ROOT\", 'path/to/your/datasets')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "def clean_data(df):\n",
    "    # remove missing values\n",
    "    df = df.dropna()\n",
    "    # remove duplicate data\n",
    "    df = df.dropDuplicates()\n",
    "    return df\n",
    "\n",
    "\n",
    "def read_data(spark, format, file_path):\n",
    "    if format==\"csv\":\n",
    "        return spark.read.format(format).load(file_path,header=True)\n",
    "    else:\n",
    "        return spark.read.format(format).load(file_path)\n",
    "\n",
    "# read sales data\n",
    "sales_df = read_data(spark, \"csv\", dataRoot+\"/sales/\")\n",
    "\n",
    "# read stock data\n",
    "stock_df = read_data(spark, \"json\", dataRoot+\"/stock/\")\n",
    "\n",
    "# read supplier data\n",
    "supplier_df = read_data(spark, \"json\", dataRoot+\"/supplier/\")\n",
    "\n",
    "# read customer data\n",
    "customer_df = read_data(spark, \"csv\", dataRoot+\"/customer/\")\n",
    "\n",
    "# read market data\n",
    "market_df = read_data(spark, \"csv\", dataRoot+\"/market/\")\n",
    "\n",
    "# read logistic data\n",
    "logistic_df = read_data(spark, \"csv\", dataRoot+\"/logistic/\")\n",
    "\n",
    "\n",
    "# data cleaning\n",
    "sales_df = clean_data(sales_df)\n",
    "stock_df = clean_data(stock_df)\n",
    "supplier_df = clean_data(supplier_df)\n",
    "customer_df = clean_data(customer_df)\n",
    "market_df = clean_data(market_df)\n",
    "logistic_df = clean_data(logistic_df)\n",
    "\n",
    "\n",
    "# convert date columns to date type\n",
    "sales_df = sales_df.withColumn(\"date_of_sale\", to_date(col(\"date_of_sale\")))\n",
    "stock_df = stock_df.withColumn(\"date_received\", to_date(col(\"date_received\")))\n",
    "supplier_df = supplier_df.withColumn(\"date_ordered\", to_date(col(\"date_ordered\")))\n",
    "\n",
    "# standardize case of string columns\n",
    "sales_df = sales_df.withColumn(\"product_name\", upper(col(\"product_name\")))\n",
    "stock_df = stock_df.withColumn(\"product_name\", upper(col(\"product_name\")))\n",
    "stock_df = stock_df.withColumn(\"location\", upper(col(\"location\")))\n",
    "supplier_df = supplier_df.withColumn(\"product_name\", upper(col(\"product_name\")))\n",
    "customer_df = customer_df.withColumn(\"customer_name\", upper(col(\"customer_name\")))\n",
    "market_df = market_df.withColumn(\"product_name\", upper(col(\"product_name\")))\n",
    "logistic_df = logistic_df.withColumn(\"product_name\", upper(col(\"product_name\")))\n",
    "\n",
    "# remove leading and trailing whitespaces\n",
    "sales_df = sales_df.withColumn(\"product_name\", trim(col(\"product_name\")))\n",
    "stock_df = stock_df.withColumn(\"location\", trim(col(\"location\")))\n",
    "\n",
    "supplier_df = supplier_df.withColumn(\"product_name\", trim(col(\"product_name\")))\n",
    "customer_df = customer_df.withColumn(\"customer_name\", trim(col(\"customer_name\")))\n",
    "market_df = market_df.withColumn(\"product_name\", trim(col(\"product_name\")))\n",
    "logistic_df = logistic_df.withColumn(\"product_name\", trim(col(\"product_name\")))\n",
    "\n",
    "# check for invalid values\n",
    "sales_df = sales_df.filter(col(\"product_name\").isNotNull())\n",
    "stock_df = stock_df.filter(col(\"location\").isNotNull())\n",
    "customer_df = customer_df.filter(col(\"gender\").isin(\"male\",\"female\"))\n",
    "market_df = market_df.filter(col(\"product_name\").isNotNull())\n",
    "logistic_df = logistic_df.filter(col(\"product_name\").isNotNull())\n",
    "\n",
    "#drop extra columns\n",
    "market_df = market_df.drop(\"price\")\n",
    "supplier_df = supplier_df.drop(\"price\")\n",
    "\n",
    "# join all data\n",
    "data_int = sales_df.join(stock_df, \"product_name\",\"leftouter\").join(supplier_df, \"product_name\",\"leftouter\").join(market_df, \"product_name\",\"leftouter\").join(logistic_df, \"product_name\",\"leftouter\").join(customer_df, \"customer_id\",\"leftouter\")  \n",
    "\n",
    "# write the cleaned data\n",
    "os.makedirs(dataRoot+\"cleaned/\", exist_ok=True)\n",
    "data_int.write.mode(\"overwrite\").format(\"parquet\").save(dataRoot+\"/cleaned/\")\n",
    "\n",
    "end = time.time()\n",
    "\n",
    "print(\"Time taken on GPU for Data Cleaning: \", end - start)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "#DO VARIOUS RETAIL DATA ANALYTICS \n",
    "\n",
    "start = time.time()\n",
    "\n",
    "# read cleaned data\n",
    "\n",
    "data = spark.read.format(\"parquet\").load(dataRoot+\"/cleaned/\")\n",
    "\n",
    "#Case when statement to create a new column to indicate whether the product is perishable or not:\n",
    "\n",
    "data = data.withColumn(\"perishable\", when(col(\"shelf_life\") <= 30, \"yes\").otherwise(\"no\"))\n",
    "\n",
    "# You can use the when() and otherwise() functions to create new columns based on certain conditions:\n",
    "\n",
    "data = data.withColumn(\"sales_status\", when(col(\"quantity_sold\") > 50, \"good\").otherwise(\"bad\"))\n",
    "\n",
    "# create a window to perform time series analysis\n",
    "window = Window.partitionBy(\"product_name\").orderBy(\"date_of_sale\")\n",
    "\n",
    "# calculate the rolling average of sales for each product\n",
    "time_series_df = data.withColumn(\"rolling_avg_sales\", avg(\"quantity_sold\").over(window))\n",
    "\n",
    "# use window function for forecasting\n",
    "\n",
    "forecast_df = time_series_df.withColumn(\"prev_sales\", lag(\"rolling_avg_sales\").over(window))\\\n",
    "    .withColumn(\"next_sales\", lead(\"rolling_avg_sales\").over(window))\n",
    "\n",
    "\n",
    "# Calculate the average price of a product, grouped by supplier\n",
    "forecast_df.groupBy(\"sup_id\").agg({\"price\": \"avg\"}).show()\n",
    "\n",
    "\n",
    "# Calculate the total quantity in stock and total sales by supplier\n",
    "forecast_df.groupBy(\"sup_id\").agg({\"quantity_in_stock\": \"sum\", \"price\": \"sum\"}).show()\n",
    "\n",
    "#Calculate the number of perishable v/s non-perishable product per location\n",
    "forecast_df.groupBy(\"perishable\").agg({\"perishable\": \"count\"}).show()\n",
    "\n",
    "\n",
    "#Calculate number of good v/s bad sales status per location\n",
    "forecast_df.groupBy(\"sales_status\").agg({\"sales_status\": \"count\"}).show()\n",
    "\n",
    "# Count the number of sales that contain a 10% off promotion\n",
    "countt = forecast_df.filter(forecast_df[\"contains_promotion\"].contains(\"10% off\")).count()\n",
    "print(countt)\n",
    "# Perform some complex analysis on the DataFrame\n",
    "\n",
    "# Calculate the total sales, quantity sold by product and location\n",
    "total_sales_by_product_location = forecast_df.groupBy(\"product_name\", \"location\").agg(sum(\"price\").alias(\"total_price\"),sum(\"quantity_ordered\").alias(\"total_quantity_sold\"),avg(\"quantity_sold\").alias(\"avg_quantity_sold\")).sort(desc(\"total_price\"))\n",
    "\n",
    "# Group the data by product_name\n",
    "grouped_df = forecast_df.groupBy(\"product_name\")\n",
    "\n",
    "#Sum the quantity_in_stock, quantity_ordered, quantity_sold, and (price * quantity_sold) for each group\n",
    "aggregated_df = grouped_df.agg(sum(\"quantity_in_stock\").alias(\"total_quantity_in_stock\"),avg(\"price\").alias(\"average_price\"),sum(\"quantity_ordered\").alias(\"total_quantity_ordered\"),sum(\"quantity_sold\").alias(\"total_quantity_sold\"),sum(col(\"price\") * col(\"quantity_sold\")).alias(\"total_sales\"),sum(\"prev_sales\").alias(\"total_prev_sales\"),sum(\"next_sales\").alias(\"total_next_sales\"),).sort(desc(\"total_sales\"))\n",
    "\n",
    "#WRITE THE AGGREGATES TO DISK\n",
    "aggregated_df.write.mode(\"overwrite\").format(\"parquet\").save(dataRoot+\"/app/data.parquet\")\n",
    "total_sales_by_product_location.write.mode(\"overwrite\").format(\"parquet\").save(dataRoot+\"/app1/data.parquet\")\n",
    "\n",
    "end = time.time()\n",
    "\n",
    "print(\"Time taken on GPU for Data Analysis: \", end - start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
